package com.doit.doit47;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Timer;
import java.util.TimerTask;

public class _08_练习之消费者之hashset {

    public static void main(String[] args) throws SQLException {
        KafkaConsumer<String, String> consumer = MyKafkaUtils.getConsumer("aaa", false, "earliest");
        HashSet<Integer> set = new HashSet<>();

        consumer.subscribe(Arrays.asList("event-log"));
        new Thread(new MyThread_HashSet(consumer,set)).start();

        Timer timer = new Timer();
        timer.schedule(new MyTask_HashSet(set),10000,5000);
    }
}

class MyThread_HashSet implements Runnable{
    private KafkaConsumer<String, String> consumer;
    private HashSet<Integer> set;
    public MyThread_HashSet(KafkaConsumer<String, String> consumer,HashSet<Integer> set){
        this.consumer = consumer;
        this.set = set;
    }
    @Override
    public void run() {
        while (true){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
            for (ConsumerRecord<String, String> record : poll) {
                String json = record.value();
                EventLog eventLog = JSON.parseObject(json, EventLog.class);
                int guid = eventLog.getGuid();
                set.add(guid);
            }
        }
    }
}

class MyTask_HashSet extends TimerTask{
    private HashSet<Integer> set;
    public MyTask_HashSet(HashSet<Integer> set){
        this.set = set;
    }
    @Override
    public void run() {
        System.out.println("当前的uv是："+set.size());
    }
}
